package com.wdl.ad.app

import com.wdl.ad.bean.{AdClickEvent, BlackListWarning}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

class FilterBlackList(maxClickCount: Long) extends KeyedProcessFunction[(Long, Long), AdClickEvent, AdClickEvent] {

  /** 定义状态 */
  lazy val countState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("count", classOf[Long]))

  /** 标识位，用来表示用户是否已经在黑名单中 */
  lazy val isSentState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("is-sent", classOf[Boolean]))

  override def processElement(value: AdClickEvent,
                              ctx: KeyedProcessFunction[(Long, Long), AdClickEvent, AdClickEvent]#Context,
                              out: Collector[AdClickEvent]): Unit = {
    // 取出状态数据
    val curCount: Long = countState.value()

    // 如果是第一条数据，那么注册第二天 0 点的定时器，用于清空状态
    if (curCount == 0){
      val ts: Long = (ctx.timerService().currentProcessingTime() / (1000 * 60 * 60 * 24) + 1) * (1000 * 60 * 60 * 24)
      ctx.timerService().registerProcessingTimeTimer(ts)
    }

    // 判断 count 值是否达到上限，如果达到，并且之前没有输出过报警信息，那么报警
    if (curCount > maxClickCount){
      if (!isSentState.value()){
        ctx.output(new OutputTag[BlackListWarning]("blackList"), BlackListWarning(value.userId.toString, value.adId.toString, "用户点击达到上限" + maxClickCount + "次！"))
        isSentState.update(true)
      }
      return
    }

    // count 值加 1
    countState.update(curCount + 1)

    out.collect(value)
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(Long, Long), AdClickEvent, AdClickEvent]#OnTimerContext, out: Collector[AdClickEvent]): Unit = {
    countState.clear()
    isSentState.clear()
  }
}
